package com.aiwiown.snackmq.broker.handler;

import com.aiwiown.snackmq.broker.auth.AuthorizationService;
import com.aiwiown.snackmq.cluster.metadata.MetadataManager;
import com.aiwiown.snackmq.cluster.raft.closeure.SnackMQApplyClosure;
import com.aiwiown.snackmq.common.auth.Action;
import com.aiwiown.snackmq.common.exception.AuthorizationException;
import com.aiwiown.snackmq.common.handler.MessageHandler;
import com.aiwiown.snackmq.common.handler.RequestContext;
import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.message.MessageStatus;
import com.aiwiown.snackmq.common.message.MessageType;
import lombok.extern.slf4j.Slf4j;

/**
 * 创建主题处理器，处理创建新主题的请求。
 * 当收到创建主题的请求时，此处理器会检查当前节点是否为 leader，
 * 如果是，则会将创建主题的任务提交给 Raft 算法处理；
 * 如果不是，则向客户端返回失败响应，并提供 leader 的地址以便重试。
 */
@Slf4j
public class CreateTopicHandler implements MessageHandler {

    // 元数据管理器，用于管理集群的元数据信息
    private final MetadataManager metadataManager;
    private final AuthorizationService authorizationService;

    /**
     * 创建主题处理器的构造函数。
     *
     * @param metadataManager 元数据管理器实例，用于处理主题创建请求。
     */
    public CreateTopicHandler(MetadataManager metadataManager,AuthorizationService authorizationService) {
        this.metadataManager = metadataManager;
        this.authorizationService = authorizationService;
    }

    /**
     * 处理创建主题的请求。
     * 如果当前节点不是 leader，则返回失败响应并提供 leader 的地址；
     * 如果是 leader，则将创建主题的任务提交给 Raft 算法处理。
     *
     * @param ctx 请求上下文，用于发送响应。
     * @param message 创建主题的请求消息，包含主题名称、分区数和副本因子等信息。
     */
    @Override
    public void handle(RequestContext ctx, Message message) {
        if (!metadataManager.isLeader()) {
            // --- 【核心修复】: 使用 Builder 模式正确构建响应 ---
            String leaderAddress = metadataManager.getLeaderAddress();
            Message.MessageBuilder responseBuilder = Message.builder()
                    .messageId(message.getMessageId())
                    .type(MessageType.RESPONSE)
                    .status(MessageStatus.REDIRECT)
                    .errorMessage("Not a leader.");

            // 仅在 leaderAddress 存在时，才将其添加到 Builder 中
            if (leaderAddress != null) {
                responseBuilder.property("redirectAddress", leaderAddress);
            }
            // 在所有配置完成后，最后调用 build() 并发送
            ctx.writeResponse(responseBuilder.build());
            return;
        }
        try {
            // 创建 SnackMQApplyClosure 实例，用于处理 Raft 算法应用结果
            SnackMQApplyClosure closure = new SnackMQApplyClosure(message, ctx, null);
            // 从消息中提取主题名称、分区数和副本因子
            String topicName = message.getProperty("topicName");
            // 【修复】授权检查：创建主题是集群级管理操作，需要对 CLUSTER 资源有 MANAGE 权限
            authorizationService.authorize(message.getProperty("authToken"), Action.MANAGE, new AuthorizationService.Resource("CLUSTER", "*"));


            int partitions = Integer.parseInt(message.getProperty("partitions", "1"));
            int replicationFactor = Integer.parseInt(message.getProperty("replicationFactor", "1"));

            // 记录创建主题的日志
            log.debug("Proposing CREATE_TOPIC: topic='{}', partitions={}, replicationFactor={}",
                    topicName, partitions, replicationFactor);
            // 提交创建主题的任务给 Raft 算法处理
            metadataManager.createTopic(topicName, partitions, replicationFactor, closure);
        } catch (AuthorizationException e) {
            log.warn("Authorization failed for create topic request: {}", e.getMessage());
            ctx.writeResponse(Message.builder().messageId(message.getMessageId()).status(MessageStatus.FAILED).errorMessage("Permission denied: " + e.getMessage()).build());
        } catch (Exception e) {
            log.error("Error creating topic", e);
            ctx.writeResponse(Message.builder().messageId(message.getMessageId()).status(MessageStatus.FAILED).errorMessage("Internal server error: " + e.getMessage()).build());
        }

    }

    /**
     * 获取此处理器处理的消息类型。
     *
     * @return 消息类型为创建主题请求。
     */
    @Override
    public MessageType getMessageType() {
        return MessageType.CREATE_TOPIC_REQUEST;
    }
}
